#include "sys.h"
#include "net.h"
#include "log.h"
#include "us_ctl.h"
#include "uentry.h"
#include "hash_table.h"
#include "thread_pool.h"
#include "udp_speed.h"

#define THREAD_COUNT_MAX 100
#define HASH_TABLE_MAX_SIZE 1000
static void send_multi(int fd, struct sockaddr_in* addr, int addr_len, char* buf, int len, int count)
{
    while (count--) {
        usleep(30000);
        sendto(fd, buf, len, 0, (struct sockaddr*)addr, addr_len);
    }
}

static int _cmp_uentry(void* a, void* b)
{
    uentry_t *ue_a, *ue_b;

    ue_a = (uentry_t*)a;
    ue_b = (uentry_t*)b;

    return uentry_cmp(ue_a, ue_b);
}
static uint16_t parse_port_range(char* portRange, uint16_t* min, uint16_t* max)
{
    char port_range[128] = { 0 };
    char *delim = ":", *p = NULL;
    uint16_t tmp;

    strcpy(port_range, portRange);
    p = strtok(port_range, delim);
    *min = strtoul(p, NULL, 10);
    p = strtok(NULL, delim);
    *max = strtoul(p, NULL, 10);

    if (*max < *min) {
        tmp = *max;
        *max = *min;
        *min = tmp;
    }
    return *min;
}
static void _send_redirect(int ctl_fd, struct sockaddr_in* peer_addr, uentry_t* ue)
{
    uctl_conf_t uconf;
    char snd_buf[UDP_BUFFER_MAX] = { 0 };

    bzero(&uconf, sizeof(uconf));
    strcpy(uconf.cmd, UCTL_CONF_CMD_REDIRECT);
    uconf.port = ue->port;
    uctl_conf_struct2str(&uconf, snd_buf, sizeof(snd_buf));
    log_trace("send redirect: %s", snd_buf);
    send_multi(ctl_fd, peer_addr, sizeof(struct sockaddr_in), snd_buf, strlen(snd_buf), 1);
}
static int is_check_pmtu(char *buf){
    return strcmp(buf, UCTL_CONF_CMD_ECHO) == 0;
}
void us_svr_run(udpspeed_svr_conf_t *svr_conf)
{
    int ret, ctl_fd;
    char ctl_buf[UDP_BUFFER_MAX] = { 0 }, snd_buf[UDP_BUFFER_MAX] = { 0 };
    fd_set rdfds;
    struct timeval tmo;
    struct sockaddr_in ctl_addr, peer_addr;
    socklen_t peer_addr_len;
    uctl_conf_t uconf;
    uentry_t* ue;
    thread_pool_t* tpool;
    uint16_t cur_port, min_port, max_port;
    struct hashTable* htable = NULL;

    cur_port = parse_port_range(svr_conf->port_range, &min_port, &max_port);
    log_trace("port_range:(%s), min:%d, max:%d", svr_conf->port_range, min_port, max_port);

    tpool = thread_pool_new(1);
    thread_pool_set_auto_extern(tpool);
    thread_pool_set_limit_number(tpool, svr_conf->max_threads ? svr_conf->max_threads : THREAD_COUNT_MAX);

    htable = hashTable_new(HASH_TABLE_MAX_SIZE);

    ctl_fd = socket_dgram_new();
    sockaddr_set(&ctl_addr, AF_INET, htons(svr_conf->ctl_port), "0.0.0.0");
    if (0 > bind(ctl_fd, (struct sockaddr*)&ctl_addr, sizeof(ctl_addr))) {
        log_syserr("ctl bind error.");
        return;
    }

    while (1) {
        bzero(ctl_buf, sizeof(ctl_buf));
        bzero(snd_buf, sizeof(snd_buf));
        FD_ZERO(&rdfds);
        FD_SET(ctl_fd, &rdfds);
        tmo.tv_sec = 1;
        tmo.tv_usec = 0;
        ret = select(ctl_fd + 1, &rdfds, NULL, NULL, &tmo);
        if (FD_ISSET(ctl_fd, &rdfds)) {

            peer_addr_len = sizeof(peer_addr);
            ret = recvfrom(ctl_fd, ctl_buf, sizeof(ctl_buf), 0, (struct sockaddr*)&peer_addr, &peer_addr_len);
            log_trace("recvfrom ret:%d", ret);
            if (ret > 0) {
                if(is_check_pmtu(ctl_buf)){
                    log_trace("check pmtu ret:%d", ret);
                    send_multi(ctl_fd, &peer_addr, sizeof(peer_addr), ctl_buf, ret, 1);
                    continue;
                }
                log_trace("Peer %s:%d, ctl conf:%s", inet_ntoa(peer_addr.sin_addr), ntohs(peer_addr.sin_port), ctl_buf);
                if (hashTable_find(htable, uentry_hash_key(&peer_addr), (void*)ue, _cmp_uentry) != NULL) {
                    log_trace("Find recode %s:%d ", inet_ntoa(peer_addr.sin_addr), ntohs(peer_addr.sin_port));
                    _send_redirect(ctl_fd, &peer_addr, ue);
                    continue;
                }
                if ((ue = uentry_new())) {
                    if (uentry_bind(ue, cur_port, min_port, max_port)) {
                        cur_port = ue->port;
                        log_trace("UEntry bind to port %d", ue->port);
                        uctl_conf_str2struct(&uconf, ctl_buf, ret);
                        if (strcmp(uconf.cmd, UCTL_CONF_CMD_REQ) == 0) {

                            memcpy(&ue->ctl_addr, &peer_addr, sizeof(struct sockaddr_in));

                            _send_redirect(ctl_fd, &peer_addr, ue);

                            ue->htable = htable;
                            ue->hnode = hashTable_add(htable, uentry_hash_key(&ue->ctl_addr), (void*)ue);

                            thread_pool_add_task(tpool, uentry_run, (void*)ue);
                        }
                    }
                }
            }
        }
    }
}
